Python win32file.error方法代码示例 |
您所在的位置：网站首页 › win32file readfile overlapped › Python win32file.error方法代码示例 |
|
本文整理汇总了Python中win32file.error方法的典型用法代码示例。如果您正苦于以下问题：Python win32file.error方法的具体用法？Python win32file.error怎么用？Python win32file.error使用的例子？那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在类win32file的用法示例。 在下文中一共展示了win32file.error方法的15个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于系统推荐出更棒的Python代码示例。 示例1: testSimpleFiles # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def testSimpleFiles(self): try: fd, filename = tempfile.mkstemp() except AttributeError: self.fail("This test requires Python 2.3 or later") os.close(fd) os.unlink(filename) handle = win32file.CreateFile(filename, win32file.GENERIC_WRITE, 0, None, win32con.CREATE_NEW, 0, None) test_data = str2bytes("Hello\0there") try: win32file.WriteFile(handle, test_data) handle.Close() # Try and open for read handle = win32file.CreateFile(filename, win32file.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None) rc, data = win32file.ReadFile(handle, 1024) self.assertEquals(data, test_data) finally: handle.Close() try: os.unlink(filename) except os.error: pass # A simple test using normal read/write operations. 开发者ID:IronLanguages，项目名称:ironpython2，代码行数:26，代码来源:test_win32file.py 示例2: testCompletionPortsMultiple # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def testCompletionPortsMultiple(self): # Mainly checking that we can "associate" an existing handle. This # failed in build 203. ioport = win32file.CreateIoCompletionPort(win32file.INVALID_HANDLE_VALUE, 0, 0, 0) socks = [] for PORT in range(9123, 9125): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind(('', PORT)) sock.listen(1) socks.append(sock) new = win32file.CreateIoCompletionPort(sock.fileno(), ioport, PORT, 0) assert new is ioport for s in socks: s.close() hv = int(ioport) ioport = new = None # The handle itself should be closed now (unless we leak references!) # Check that. try: win32file.CloseHandle(hv) raise RuntimeError("Expected close to fail!") except win32file.error, details: self.failUnlessEqual(details.winerror, winerror.ERROR_INVALID_HANDLE) 开发者ID:IronLanguages，项目名称:ironpython2，代码行数:27，代码来源:test_win32file.py 示例3: _IOCPServerThread # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def _IOCPServerThread(self, handle, port, drop_overlapped_reference): overlapped = pywintypes.OVERLAPPED() win32pipe.ConnectNamedPipe(handle, overlapped) if drop_overlapped_reference: # Be naughty - the overlapped object is now dead, but # GetQueuedCompletionStatus will still find it. Our check of # reference counting should catch that error. overlapped = None # even if we fail, be sure to close the handle; prevents hangs # on Vista 64... try: self.failUnlessRaises(RuntimeError, win32file.GetQueuedCompletionStatus, port, -1) finally: handle.Close() return result = win32file.GetQueuedCompletionStatus(port, -1) ol2 = result[-1] self.failUnless(ol2 is overlapped) data = win32file.ReadFile(handle, 512)[1] win32file.WriteFile(handle, data) 开发者ID:IronLanguages，项目名称:ironpython2，代码行数:24，代码来源:test_win32file.py 示例4: testEmptyDir # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def testEmptyDir(self): test_path = os.path.join(win32api.GetTempPath(), "win32file_test_directory") try: # Note: previously used shutil.rmtree, but when looking for # reference count leaks, that function showed leaks! os.rmdir # doesn't have that problem. os.rmdir(test_path) except os.error: pass os.mkdir(test_path) try: num = 0 for i in win32file.FindFilesIterator(os.path.join(test_path, "*")): num += 1 # Expecting "." and ".." only self.failUnlessEqual(2, num) finally: os.rmdir(test_path) 开发者ID:IronLanguages，项目名称:ironpython2，代码行数:20，代码来源:test_win32file.py 示例5: test_connect_with_payload # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def test_connect_with_payload(self): giveup_event = win32event.CreateEvent(None, 0, 0, None) t = threading.Thread(target=self.connect_thread_runner, args=(True, giveup_event)) t.start() time.sleep(0.1) s2 = socket.socket() ol = pywintypes.OVERLAPPED() s2.bind(('0.0.0.0', 0)) # connectex requires the socket be bound beforehand try: win32file.ConnectEx(s2, self.addr, ol, str2bytes("some expected request")) except win32file.error, exc: win32event.SetEvent(giveup_event) if exc.winerror == 10022: # WSAEINVAL raise TestSkipped("ConnectEx is not available on this platform") raise # some error error we don't expect. 开发者ID:IronLanguages，项目名称:ironpython2，代码行数:18，代码来源:test_win32file.py 示例6: test_connect_without_payload # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def test_connect_without_payload(self): giveup_event = win32event.CreateEvent(None, 0, 0, None) t = threading.Thread(target=self.connect_thread_runner, args=(False, giveup_event)) t.start() time.sleep(0.1) s2 = socket.socket() ol = pywintypes.OVERLAPPED() s2.bind(('0.0.0.0', 0)) # connectex requires the socket be bound beforehand try: win32file.ConnectEx(s2, self.addr, ol) except win32file.error, exc: win32event.SetEvent(giveup_event) if exc.winerror == 10022: # WSAEINVAL raise TestSkipped("ConnectEx is not available on this platform") raise # some error error we don't expect. 开发者ID:IronLanguages，项目名称:ironpython2，代码行数:18，代码来源:test_win32file.py 示例7: lseek # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def lseek(self, port, pos, how): """ Use lseek on the device. The device is unseekable so PASS is returned when lseek command fails and vice versa. :param port: Name of the port :param pos: Offset :param how: Relative offset os.SEEK_{SET,CUR,END} """ fd = self._open([port])[0] try: os.lseek(fd, pos, how) except Exception as inst: if inst.errno == 29: print("PASS: the lseek failed as expected") else: print(inst) print("FAIL: unknown error") else: print("FAIL: the lseek unexpectedly passed") 开发者ID:avocado-framework，项目名称:avocado-vt，代码行数:23，代码来源:virtio_console_guest.py 示例8: close # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def close(self, filepath): """ Close open port. :param filepath: File to close. """ hFile = None path = self.ports[filepath]["path"] if path is not None: if path in list(self.files.keys()): hFile = self.files[path] del self.files[path] if hFile is not None: try: win32file.CloseHandle(hFile) except win32file.error as inst: print("FAIL: Closing the file: " + str(inst)) return print("PASS: Close") 开发者ID:avocado-framework，项目名称:avocado-vt，代码行数:21，代码来源:virtio_console_guest.py 示例9: open # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def open(self, name): """ Direct open devices. :param name: Port name. :return: 0 on success """ path = self.ports[name]['path'] try: self.files[path] = win32file.CreateFile(path, win32file.GENERIC_WRITE | win32file.GENERIC_READ, 0, None, win32file.OPEN_EXISTING, win32file.FILE_ATTRIBUTE_NORMAL, None) except win32file.error as exc_detail: print("%s\nFAIL: Failed open file %s" % (str(exc_detail), name)) return exc_detail print("PASS: All files opened correctly.") 开发者ID:avocado-framework，项目名称:avocado-vt，代码行数:23，代码来源:virtio_console_guest.py 示例10: blksize # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def blksize(path): """ Get optimal file system buffer size (in bytes) for I/O calls. """ diskfreespace = win32file.GetDiskFreeSpace dirname = os.path.dirname(fullpath(path)) try: cluster_sectors, sector_size = diskfreespace(dirname)[:2] size = cluster_sectors * sector_size except win32file.error as e: if e.winerror != winerror.ERROR_NOT_READY: raise sleep(3) size = blksize(dirname) return size 开发者ID:deplicate，项目名称:deplicate，代码行数:19，代码来源:nt.py 示例11: handle_cmd_response # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def handle_cmd_response(self, resp): command = self.sent_commands[resp['request_id']]['command'] del self.sent_commands[resp['request_id']] if resp['error'] != 'success': logger.error(f'Error with command {command!s}. Response: {resp!s}') return elif command[0] != 'get_property': return param = command[1] data = resp['data'] if param == 'pause': self.vars['state'] = 1 if data else 2 if param in self.WATCHED_PROPS: self.vars[param] = data self.updated_props_count += 1 if self.updated_props_count == len(self.WATCHED_PROPS): self.update_status() 开发者ID:iamkroot，项目名称:trakt-scrobbler，代码行数:19，代码来源:mpv.py 示例12: testSimpleOverlapped # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def testSimpleOverlapped(self): # Create a file in the %TEMP% directory. import win32event testName = os.path.join( win32api.GetTempPath(), "win32filetest.dat" ) desiredAccess = win32file.GENERIC_WRITE overlapped = pywintypes.OVERLAPPED() evt = win32event.CreateEvent(None, 0, 0, None) overlapped.hEvent = evt # Create the file and write shit-loads of data to it. h = win32file.CreateFile( testName, desiredAccess, 0, None, win32file.CREATE_ALWAYS, 0, 0) chunk_data = str2bytes("z") * 0x8000 num_loops = 512 expected_size = num_loops * len(chunk_data) for i in range(num_loops): win32file.WriteFile(h, chunk_data, overlapped) win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE) overlapped.Offset = overlapped.Offset + len(chunk_data) h.Close() # Now read the data back overlapped overlapped = pywintypes.OVERLAPPED() evt = win32event.CreateEvent(None, 0, 0, None) overlapped.hEvent = evt desiredAccess = win32file.GENERIC_READ h = win32file.CreateFile( testName, desiredAccess, 0, None, win32file.OPEN_EXISTING, 0, 0) buffer = win32file.AllocateReadBuffer(0xFFFF) while 1: try: hr, data = win32file.ReadFile(h, buffer, overlapped) win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE) overlapped.Offset = overlapped.Offset + len(data) if not data is buffer: self.fail("Unexpected result from ReadFile - should be the same buffer we passed it") except win32api.error: break h.Close() 开发者ID:IronLanguages，项目名称:ironpython2，代码行数:37，代码来源:test_win32file.py 示例13: testCompletionPortsNonQueued # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def testCompletionPortsNonQueued(self, test_overlapped_death = 0): # In 204 we had a reference count bug when OVERLAPPED objects were # associated with a completion port other than via # PostQueuedCompletionStatus. This test is based on the reproduction # reported with that bug. # Create the pipe. BUFSIZE = 512 pipe_name = r"\\.\pipe\pywin32_test_pipe" handle = win32pipe.CreateNamedPipe(pipe_name, win32pipe.PIPE_ACCESS_DUPLEX| win32file.FILE_FLAG_OVERLAPPED, win32pipe.PIPE_TYPE_MESSAGE| win32pipe.PIPE_READMODE_MESSAGE| win32pipe.PIPE_WAIT, 1, BUFSIZE, BUFSIZE, win32pipe.NMPWAIT_WAIT_FOREVER, None) # Create an IOCP and associate it with the handle. port = win32file.CreateIoCompletionPort(-1, 0, 0, 0) win32file.CreateIoCompletionPort(handle, port, 1, 0) t = threading.Thread(target=self._IOCPServerThread, args=(handle,port, test_overlapped_death)) t.setDaemon(True) # avoid hanging entire test suite on failure. t.start() try: time.sleep(0.1) # let thread do its thing. try: win32pipe.CallNamedPipe(r"\\.\pipe\pywin32_test_pipe", str2bytes("Hello there"), BUFSIZE, 0) except win32pipe.error: # Testing for overlapped death causes this if not test_overlapped_death: raise finally: if not test_overlapped_death: handle.Close() t.join(3) self.failIf(t.isAlive(), "thread didn't finish") 开发者ID:IronLanguages，项目名称:ironpython2，代码行数:39，代码来源:test_win32file.py 示例14: testEncrypt # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def testEncrypt(self): fname = tempfile.mktemp("win32file_test") f = open(fname, "wb") f.write(str2bytes("hello")) f.close() f = None try: try: win32file.EncryptFile(fname) except win32file.error, details: if details.winerror != winerror.ERROR_ACCESS_DENIED: raise print "It appears this is not NTFS - cant encrypt/decrypt" win32file.DecryptFile(fname) 开发者ID:IronLanguages，项目名称:ironpython2，代码行数:16，代码来源:test_win32file.py 示例15: conn_loop # 需要导入模块: import win32file [as 别名] # 或者: from win32file import error [as 别名] def conn_loop(self): self.is_running = True self.update_vars() self.file_handle = win32file.CreateFile( self.ipc_path, win32file.GENERIC_READ | win32file.GENERIC_WRITE, 0, None, win32file.OPEN_EXISTING, 0, None ) while self.is_running: try: while not self.write_queue.empty(): win32file.WriteFile( self.file_handle, self.write_queue.get_nowait()) except win32file.error: logger.debug('Exception while writing to pipe.', exc_info=True) self.is_running = False break size = win32file.GetFileSize(self.file_handle) if size > 0: while size > 0: # pipe has data to read _, data = win32file.ReadFile(self.file_handle, 4096) self.on_data(data) size = win32file.GetFileSize(self.file_handle) else: time.sleep(1) win32file.CloseHandle(self.file_handle) logger.debug('Pipe closed.') 开发者ID:iamkroot，项目名称:trakt-scrobbler，代码行数:32，代码来源:mpv.py注：本文中的win32file.error方法示例由纯净天空整理自Github/MSDocs等开源代码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
| CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3 |